package consumer

import (
	conf "gitlab.sensechar.com/kafkaGo/config"
	"github.com/Shopify/sarama"
	cluster "github.com/bsm/sarama-cluster"
	"strings"
	"time"
)

const topicSign = "kafka-go"

// sarama 与 sarama-cluster区别是 sarama不支持消费者组
type KafkaConsumerClient struct {
	// 集群节点
	BrokerServers []string
	// 消费者配置文件
	Config *sarama.Config
	// 集群消费者配置文件
	ClusterConf *cluster.Config
	// 普通消费者
	RawConsumer sarama.Consumer
	// 集群消费者
	ClusterConsumer *cluster.Consumer
	// 消费者组id
	GoupID string
	// 集群主题
	Topics []string
}

// 没有加入消费者组的消费者(类似于发布订阅)
func NewConsumer(option *conf.Option) *KafkaConsumerClient {
	config := sarama.NewConfig()
	// 版本
	config.Version = sarama.V0_11_0_2
	// 从当前的偏移量开始消费
	config.Consumer.Offsets.Initial = sarama.OffsetNewest

	// 设置自动提交
	config.Consumer.Offsets.AutoCommit.Enable = true
	config.Consumer.Offsets.AutoCommit.Interval = 1*time.Second

	return &KafkaConsumerClient{
		BrokerServers:  option.Brokers,
		Config: config,
	}
}

func (k *KafkaConsumerClient) Consumer() (err error) {
	k.RawConsumer,err=sarama.NewConsumer(k.BrokerServers,k.Config)
	if err!=nil{
		return
	}
	return
}

func (k *KafkaConsumerClient) TopicsAndPartitions() (map[string]int,error) {
	topics,err:= k.RawConsumer.Topics()
	if err!=nil{
		return nil,err
	}
	topicMap :=make(map[string]int)
	for _,v := range topics{
		if strings.HasPrefix(v, topicSign) {
			if p, e := k.RawConsumer.Partitions(v); e == nil {
				topicMap[v] = len(p)
			}
		}
	}
	return topicMap,nil
}


func (k *KafkaConsumerClient) ClustConsumer() (err error){
	k.ClusterConsumer,err = cluster.NewConsumer(k.BrokerServers,k.GoupID,k.Topics,k.ClusterConf)
	if err!=nil{
		return
	}
	return
}

// 配置说明 https://www.cnblogs.com/yx88/p/11013338.html
func NewClusterConsumer(option *conf.Option) *KafkaConsumerClient {
	config := cluster.NewConfig()
	// 版本
	config.Version = sarama.V0_11_0_2
    // 从当前的偏移量开始消费
	config.Consumer.Offsets.Initial = sarama.OffsetNewest

	// 是否等待成功和失败后的响应
	config.Group.Return.Notifications = true
	config.Consumer.Return.Errors = true

	// 设置自动提交
	config.Consumer.Offsets.AutoCommit.Enable = true
	config.Consumer.Offsets.AutoCommit.Interval = 1*time.Second

	// 事务设置
	//config.Consumer.IsolationLevel=sarama.ReadUncommitted

	kafkaClient := &KafkaConsumerClient{
		BrokerServers:   option.Brokers,
		ClusterConf:     config,
		GoupID:          option.GroupID,
	}
	return kafkaClient
}

// https://gitee.com/johng/gf/blob/83f14a102f99a75f91e0996379934fe8fae93e77/g/database/gkafka/gkafka.go

